package com.yanggu.bigdata.realtime.app.function

import cn.hutool.core.util.StrUtil
import com.alibaba.fastjson.JSONObject
import com.yanggu.bigdata.realtime.common.GmallConfig
import com.yanggu.bigdata.realtime.utils.RedisUtil
import org.apache.commons.lang3.StringUtils
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}

import java.sql.{Connection, DriverManager, PreparedStatement}

/**
 * 维度表数据直接保存到Hbase中, 同时如果有更新数据时删除redis中的缓存
 */
class HbaseDimRichSinkFunction extends RichSinkFunction[JSONObject]{

  /**
   * 使用懒加载的方式初始化connection
   */
  private lazy val connection: Connection = {
    Class.forName(GmallConfig.PHOENIX_DRIVER)
    val tempConnection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER)
    //Phoenix必须显示设定自动提交, Phoenix默认不提交数据
    tempConnection.setAutoCommit(true)
    tempConnection
  }

  //SQL：upsert into db.tn(id,tm_name,aa,bb) values('...','...','...','...')
  override def invoke(value: JSONObject, context: SinkFunction.Context) = {
    var statement: PreparedStatement = null
    try {
      val sinkTable = value.getString("sinkTable")
      //如果是更新数据, 需要删除redis中缓存的数据保持数据的一致性
      val jsonData = value.getJSONObject("after")
      if (StrUtil.equals("update", value.getString("operation"))) {
        //redis的key => database:table_name:id 都是大写
        RedisUtil.deleteByKey(s"${GmallConfig.HBASE_SCHEMA}:${sinkTable.toUpperCase}:${jsonData.getString("id")}")
      }
      val sql = s"UPSERT INTO ${GmallConfig.HBASE_SCHEMA}.$sinkTable (${StringUtils.join(jsonData.keySet(), ",")})" +
        s" VALUES('${StringUtils.join(jsonData.values(), "', '")}')"
      statement = connection.prepareStatement(sql)
      //执行sql
      statement.executeUpdate()
    } finally {
      if (statement != null) {
        statement.close()
      }
    }
  }

}
